package com.shenmazg6.task;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.shenmazg6.mapper.TbProductMapper;
import com.shenmazg6.pojo.TbProduct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * @author 军哥
 * @version 1.0
 * @description: TaskElasticSearch
 * @date 2022/12/26 15:07
 */

@Component
@EnableScheduling
@Slf4j
public class TaskElasticSearch {

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Autowired
    TbProductMapper tbProductMapper;

    private Date syncTime = null;

    /**
     * 把产品数据从MySQL同步到ES中
     */
    @Scheduled(fixedDelay = 1000*60)
    public void syncProductToElasticSearch() {

        //-- 1检查索引是否存在
        if(!elasticsearchRestTemplate.indexOps(TbProduct.class).exists()) {

            log.info("创建索引......");

            // 创建索引
            elasticsearchRestTemplate.indexOps(TbProduct.class).create();
            // 手动创建映射
            Document mapping = elasticsearchRestTemplate.indexOps(TbProduct.class).createMapping();
            elasticsearchRestTemplate.indexOps(TbProduct.class).putMapping(mapping);

            // 重置同步时间
            syncTime = null;
        }

        //--2 全量同步
        if(syncTime == null) {

            log.info("进行全量同步......");

            syncTime = new Date(System.currentTimeMillis());

            QueryWrapper<TbProduct> wrapper = new QueryWrapper<>();
            wrapper.lambda().le(TbProduct::getUpdateTime, syncTime);

            List<TbProduct> tbProducts = tbProductMapper.selectList(wrapper);
            elasticsearchRestTemplate.save(tbProducts);

            return;
        }

        //--3 增量同步
        log.info("增量同步......");

        QueryWrapper<TbProduct> wrapper = new QueryWrapper<>();
        wrapper.lambda().gt(TbProduct::getUpdateTime, syncTime).orderByDesc(TbProduct::getUpdateTime);

        List<TbProduct> tbProducts = tbProductMapper.selectList(wrapper);

        if(tbProducts!=null && tbProducts.size()>0) {
            elasticsearchRestTemplate.save(tbProducts);

            // 更新时间标志
            syncTime = tbProducts.get(0).getUpdateTime();
        }
    }

}
